package me.tuyou.function;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.ReduceOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple3;

public class ReduceTest {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<Tuple3<Integer, String, Double>> input = env.fromElements(
                Tuple3.of(1, "a", 1.0),
                Tuple3.of(1, "a", 1.1),
                Tuple3.of(2, "b", 2.2),
                Tuple3.of(2, "b", 2.3),
                Tuple3.of(3, "c", 3.3),
                Tuple3.of(3, "c", 3.4)
        );

        UnsortedGrouping<Tuple3<Integer, String, Double>> group = input.groupBy(1);
        ReduceOperator<Tuple3<Integer, String, Double>> reduce = group.reduce((value1, value2) -> {
            value1.f0 += value2.f0;
            value1.f2 += value2.f2;
            return value1;
        });
        reduce.print();
    }
}
